/**
 * 	Copyright 2011 rkehoe
 *
 * 	Licensed under the Apache License, Version 2.0 (the "License");
 *  you may not use this file except in compliance with the License.
 * 	You may obtain a copy of the License at
 *
 *  http://www.apache.org/licenses/LICENSE-2.0
 *
 *  Unless required by applicable law or agreed to in writing, software
 *  distributed under the License is distributed on an "AS IS" BASIS,
 *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 *  See the License for the specific language governing permissions and
 *	limitations under the License.
 *
 */
package queryable.store.sql;

import java.rmi.server.UID;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Statement;
import java.util.ArrayList;
import java.util.Collection;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.UUID;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.locks.Lock;

import javax.sql.DataSource;

import org.springframework.jdbc.core.RowMapper;

import queryable.store.EntityReflector;
import queryable.store.EntityReflector.DomainObjectMetadata;
import queryable.store.AbstractStore;
import queryable.store.IDGenerator;
import queryable.store.IFieldMetadata;
import queryable.store.IStore;
import queryable.store.annotations.Index;

/**
 * Simple JDBC store.
 * 
 * @author rkehoe
 * @param <T>
 * 
 */
public class HybridHashMapStore<T> extends AbstractStore<T>
{
	private static EntityReflector	       entityReflector	     = new EntityReflector();

	private static final int	           BATCT_INSERT_SIZE	 = 1000;
	private PreparedStatement	           insertPstmt;
	private Map<String, PreparedStatement>	pStmtCache;
	private final ISQLGenerator	           sqlGenerator;
	private String	                       tableName;
	private final DataSource	           dataSource;
	private Connection	                   preparedStatementConnection;
	private final IDGenerator	           idGenerator;
	private final Map<Object, T>	       objectMap;
	private String	                       idFieldName;

	public HybridHashMapStore(String name, Class<T> domainObjectClass, IStorageProvider storageProvider, Map<Object, T> map) throws Exception
	{
		this.objectMap = map;

		this.tableName = name != null ? name : deriveName(domainObjectClass);
		this.dataSource = storageProvider.getDataSource();
		this.preparedStatementConnection = this.dataSource.getConnection();
		//TODO - autocommit????
		this.preparedStatementConnection.setAutoCommit(true);

		DomainObjectMetadata entityMetadata = entityReflector.getDomainObjectMetadata(domainObjectClass);

		List<Index> indexes = entityMetadata.getIndexes();
		List<IFieldMetadata> fields = entityMetadata.getQueryFields();

		this.idFieldName = entityMetadata.getIdFieldMetadata().getName();

		this.sqlGenerator = storageProvider.getGenerator();

		String tableDDL = sqlGenerator.getTableDDL(fields, tableName);
		System.out.println("Creating Objects: " + tableDDL);

		Statement ddlStatement = preparedStatementConnection.createStatement();
		ddlStatement.execute(tableDDL);

		for (Index index : indexes)
		{
			List<String> indexDDL = sqlGenerator.getIndexDDL(tableName, null, index.columnNames());
			for (String indxStmt : indexDDL)
			{
				System.out.println("Creating indx: " + indxStmt);
				ddlStatement.execute(indxStmt);
			}
		}

		ddlStatement.close();

		String insertDML = sqlGenerator.getInsertDML(fields, tableName);
		System.out.println("Generating Insert Statement: " + insertDML);
		this.insertPstmt = preparedStatementConnection.prepareStatement(insertDML);
		this.pStmtCache = new HashMap<String, PreparedStatement>();

		this.idGenerator = createIDGenerator(entityMetadata);
	}

	/**
	 * @param domainObjectClass
	 * @return
	 */
	private static String deriveName(Class<?> domainObjectClass)
	{
		DomainObjectMetadata entityMetadata = entityReflector.getDomainObjectMetadata(domainObjectClass);

		if (entityMetadata.getTable() != null)
		{
			String storeName = entityMetadata.getTable().name();
			return storeName;
		}
		else
		{
			String storeName = domainObjectClass.getSimpleName();
			return storeName;
		}
	}

	/**
	 * Create an ID generator based on the Type of the @Id field - currently
	 * only longs and strings are handled.
	 * 
	 * @param metadata
	 * @return
	 */
	private IDGenerator createIDGenerator(DomainObjectMetadata md)
	{
		Class<? extends Object> idType = md.getIdFieldMetadata().getType();
		if (idType == Long.class)
			return new SimpleLongIncrementor();
		if (idType == Integer.class)
			return new SimpleIntegerIncrementor();
		else
			return new UIDGenerator();
	}

	/*
	 * (non-Javadoc)
	 * 
	 * @see queryable.store.IDAO#insert(T)
	 */
	@Override
	public void add(T entity)
	{
		this.writeLock.lock();
		try
		{
			Object id = entityReflector.setIDifNull(this.idGenerator, entity);
			Map<String, Object> properties = entityReflector.getProperties(entity);
			int ii = 1;
			for (Entry<String, Object> property : properties.entrySet())
			{
				Object value = property.getValue();
				insertPstmt.setObject(ii, value);
				ii++;
			}

			this.objectMap.put(id, entity);

			insertPstmt.executeUpdate();
		}
		catch (Exception e)
		{
			throw new RuntimeException(e);
		}
		finally
		{
			this.writeLock.unlock();
		}
	}

	/*
	 * (non-Javadoc)
	 * 
	 * @see queryable.store.IDAO#insert(java.util.Collection)
	 */
	@Override
	public void addAll(Collection<T> list)
	{
		this.writeLock.lock();
		try
		{
			insertPstmt.clearBatch();
			int jj = 1;
			for (T entity : list)
			{
				Object id = entityReflector.setIDifNull(this.idGenerator, entity);
				Map<String, Object> properties = entityReflector.getProperties(entity);

				int ii = 1;
				for (Entry<String, Object> entry : properties.entrySet())
				{
					Object value = entry.getValue();
					insertPstmt.setObject(ii, value);
					ii++;
				}

				this.objectMap.put(id, entity);

				insertPstmt.addBatch();

				jj++;

				if (jj % BATCT_INSERT_SIZE == 0)
					insertPstmt.executeBatch();
			}
			insertPstmt.executeBatch();
		}
		catch (Exception e)
		{
			throw new RuntimeException(e);
		}
		finally
		{
			this.writeLock.unlock();
		}
	}

	/*
	 * (non-Javadoc)
	 * 
	 * @see queryable.store.IStore#clear()
	 */
	@Override
	public void clear()
	{
		this.writeLock.lock();
		try
		{
			this.executeUpdate("truncate table " + this.tableName, new Object[0]);
			this.objectMap.clear();
		}
		catch (Exception e)
		{
			throw new RuntimeException(e);
		}
		finally
		{
			this.writeLock.unlock();
		}
	}

	/*
	 * (non-Javadoc)
	 * 
	 * @see queryable.store.IStore#remove(java.lang.String, java.lang.Object[])
	 */
	@Override
	public int remove(String predicate, Object[] args)
	{
		this.writeLock.lock();
		try
		{
			// Step 1. 'Remove' items from the map
			RowMapper<T> rowMapper = new RowMapper<T>()
			{
				@Override
				public T mapRow(ResultSet rs, int row) throws SQLException
				{
					Object key = rs.getObject(1);
					return objectMap.remove(key);
				}
			};

			List<T> removedObjects = this.select(this.idFieldName, predicate, args, rowMapper);

			// Step 2. Delete rows which match predicate
			String query = "delete from " + this.tableName + " where " + predicate;
			int rowCount = this.executeUpdate(query, args);
			
			assert (rowCount == removedObjects.size());

			return rowCount;
		}
		catch (Exception e)
		{
			throw new RuntimeException(e);
		}
		finally
		{
			this.writeLock.unlock();
		}
	}

	/*
	 * (non-Javadoc)
	 * 
	 * @see queryable.store.IDAO#size()
	 */
	@Override
	public int size()
	{
		return this.objectMap.size();
	}

	private int executeUpdate(String statement, Object[] args) throws Exception
	{
		if (args == null)
			args = new Object[0];
		PreparedStatement pstmt = this.pStmtCache.get(statement);
		if (pstmt == null)
		{
			pstmt = this.preparedStatementConnection.prepareStatement(statement);
			this.pStmtCache.put(statement, pstmt);
			System.out.println("Creating PreparedStatement: " + pstmt);
		}
		pstmt.clearBatch();
		for (int ii = 0; ii < args.length; ii++)
		{
			pstmt.setObject(ii + 1, args[ii]);
		}
		return pstmt.executeUpdate();
	}

	/*
	 * (non-Javadoc)
	 * 
	 * @see queryable.store.IDAO#select(java.lang.String, java.lang.String,
	 * java.lang.Object[], org.springframework.jdbc.core.RowMapper)
	 */
	@Override
	public <R> List<R> select(String fields, String predicate, Object[] args, RowMapper<R> rowMapper)
	{
		String query = "select " + fields + " from " + this.tableName;
		if (predicate != null && !predicate.isEmpty())
			query = query + " where " + predicate;
		try
		{
			List<R> list = new ArrayList<R>();
			ResultSet resultSet = this.execute(query, args);
			int ii = 0;
			while (resultSet.next())
			{
				R mappedRow = rowMapper.mapRow(resultSet, ++ii);
				if(mappedRow!=null)
					list.add(mappedRow);
			}
			return list;
		}
		catch (Exception e)
		{
			e.printStackTrace();
			throw new RuntimeException(e);
		}
	}

	/*
	 * (non-Javadoc)
	 * 
	 * @see queryable.store.IDAO#selectWhere(java.lang.String,
	 * java.lang.Object[])
	 */
	@Override
	public List<T> select(String predicate, Object[] args)
	{
		RowMapper<T> rowMapper = new RowMapper<T>()
		{
			@Override
			public T mapRow(ResultSet rs, int row) throws SQLException
			{
				Object key = rs.getObject(1);
				return objectMap.get(key);
			}
		};

		List<T> select = this.select(this.idFieldName, predicate, args, rowMapper);
		return select;
	}

	/*
	 * Executes a statement as prepared statement - caching if necessary.
	 */
	private ResultSet execute(String sql, Object[] args) throws Exception
	{
		if (args == null)
			args = new Object[0];
		PreparedStatement pstmt = this.pStmtCache.get(sql);
		if (pstmt == null)
		{
			pstmt = this.preparedStatementConnection.prepareStatement(sql);
			this.pStmtCache.put(sql, pstmt);
			System.out.println("Creating PreparedStatement: " + pstmt);
		}
		pstmt.clearBatch();
		for (int ii = 0; ii < args.length; ii++)
		{
			pstmt.setObject(ii + 1, args[ii]);
		}
		return pstmt.executeQuery();
	}

	/*
	 * (non-Javadoc)
	 * 
	 * @see queryable.store.IStore#isEmpty()
	 */
	@Override
	public boolean isEmpty()
	{
		return this.size() == 0;
	}

	/*
	 * (non-Javadoc)
	 * 
	 * @see queryable.store.IStore#iterator()
	 */
	@Override
	public Iterator<T> iterator()
	{
		return this.objectMap.values().iterator();
	}
}
